package com.sheepone.distributor.expand;

import com.sheepone.distributor.DisruptorHolder;
import com.sheepone.distributor.Distributor;
import com.sheepone.distributor.WorkerStation;
import com.sheepone.distributor.WorkerStationManager;
import com.sheepone.distributor.listener.DistributorAwareListener;
import com.sheepone.distributor.listener.DistributorListenerAdapter;
import com.sheepone.util.console.table.Cell;
import com.sheepone.util.console.table.ConsoleTableUtil;
import com.sheepone.util.enums.Locate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Miss.杨
 * @description
 * @since 2024/1/26 - 14:00
 */
public class DataMonitor extends DistributorListenerAdapter implements Runnable, DistributorAwareListener {
    private static final Logger logger = LoggerFactory.getLogger(new Object(){}.getClass().getEnclosingClass());
    private final long intervalTime;

    private AtomicBoolean stopped = new AtomicBoolean(false);

    private final Thread thread = new Thread(this);
    private  Distributor distributor;

    public DataMonitor(long intervalTime) {
        this.intervalTime = intervalTime;
    }

    @Override
    public void afterStarting() {
        thread.start();
    }

    @Override
    public void beforeStopping() {
        stopped.compareAndSet(false, true);
        thread.interrupt();
    }

    @Override
    public void setDistributor(Distributor distributor) {
        this.distributor=distributor;
    }

    @Override
    public synchronized void run() {
        List<WorkerStation> stations = WorkerStationManager.INSTANCE.getWorkerStations().values().stream().toList();
        List<Cell> headers = new ArrayList<>();
        headers.add(new Cell(Locate.CENTER, "Disruptor"));
        headers.add(new Cell(Locate.CENTER, this.distributor.getName()));
        headers.addAll(stations.stream().map(station -> new Cell(Locate.CENTER, station.getName())).toList());
        List<Cell> nums = new ArrayList<>();
        for (int i = 0; i < headers.size(); i++) {
            nums.add(new Cell(Locate.CENTER));
        }
        nums.get(0).setValue("HasPublishedNum");
        while (!stopped.get()) {
            try {
                TimeUnit.SECONDS.sleep(intervalTime);
                nums.get(1).setValue(this.distributor.hasPublishNum());
                List<Long> cells = stations.stream().map(DisruptorHolder::hasPublishedNum).toList();
                for (int i = 0; i < cells.size(); i++) {
                    nums.get(i + 2).setValue(cells.get(i));
                }
                ConsoleTableUtil.newTable().addHeaders(headers).addRow(nums).print();
            } catch (InterruptedException e) {
                if(logger.isDebugEnabled()){
                    logger.debug("DataMonitor thread interrupted");
                }
            }
        }
    }
}